package iotdata

import (
	"context"
	"devops-super/api"
	v1 "devops-super/api/iotdata/v1"
	"devops-super/internal/consts"
	"devops-super/internal/consts/cacheConsts"
	"devops-super/internal/dao"
	"devops-super/internal/model/entity"
	"devops-super/internal/model/mymodel"
	"devops-super/internal/service"
	"devops-super/utility/util/arrayUtil"
	"devops-super/utility/util/dbUtil"
	"devops-super/utility/util/minioUtil"
	"devops-super/utility/util/waveFileSave"
	"encoding/hex"
	"errors"
	"github.com/gogf/gf/v2/database/gdb"
	"github.com/gogf/gf/v2/errors/gcode"
	"github.com/gogf/gf/v2/errors/gerror"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/os/gtime"
	"github.com/gogf/gf/v2/util/gconv"
	"time"
	"unicode"
)

type sIotdata struct{}

func init() {
	service.RegisterIotdata(New())
}

func New() *sIotdata {
	return &sIotdata{}
}

var maxTotalSearch = 3000000

// GetPageLst 分页获取物联网数据
func (s *sIotdata) GetPageLst(ctx context.Context, req *v1.IotDataGetPageLstReq) (res *v1.IotDataGetPageLstRes, err error) {
	res = &v1.IotDataGetPageLstRes{
		PageLstRes: &api.PageLstRes[any]{
			List:  make([]any, 0),
			Total: 0,
		},
	}
	var (
		order              = "sampletime desc"
		allTables          []*mymodel.DistributeTableResponse
		deviceNoTableField = dao.IotAeParamDemo.Columns().DeviceNo
		startTime          time.Time
		endTime            time.Time
	)
	if req.OrderBy != "" {
		order = req.OrderBy
	}

	// 参数校验增强
	if len(req.DeviceNos) == 0 {
		return nil, gerror.NewCode(gcode.CodeInvalidParameter, "deviceNo required")
	}
	deviceInfo0, err := service.Device().GetByDeviceNo(ctx, req.DeviceNos[0])
	if err != nil || deviceInfo0 == nil || deviceInfo0.Id == 0 {
		return nil, gerror.NewCode(gcode.CodeInvalidParameter, "deviceNo not exist")
	}
	isSuperDept, currentUser, _ := service.User().IsCurrentUserSuperDept(ctx)
	if !isSuperDept && currentUser.DeptId != int(deviceInfo0.DeptId) {
		return nil, gerror.NewCode(gcode.CodeSecurityReason, "permission deny")
	}

	productInfo, err := service.Product().GetByProductId(ctx, deviceInfo0.ProductId)
	if err != nil || productInfo == nil || productInfo.Id == 0 {
		return nil, gerror.NewCode(gcode.CodeInvalidParameter, "productInfo not exist")
	}
	productKey := productInfo.ProductKey
	// 时间条件处理优化
	switch req.TimeType {
	case 0:
		if len(req.TimeRange) != 2 {
			return nil, gerror.NewCode(gcode.CodeInvalidParameter, "时间范围参数格式错误")
		}
		// 新增时间转换逻辑
		startTime, err = time.Parse(time.RFC3339, req.TimeRange[0])
		if err != nil {
			return nil, gerror.WrapCode(gcode.CodeInvalidParameter, err, "开始时间格式错误")
		}
		endTime, err = time.Parse(time.RFC3339, req.TimeRange[1])
		if err != nil {
			return nil, gerror.WrapCode(gcode.CodeInvalidParameter, err, "结束时间格式错误")
		}
	case 1:
		startTime = gtime.Now().Add(-time.Duration(req.Recent) * time.Second).Time
		endTime = gtime.Now().Time
	default:
		return nil, gerror.NewCode(gcode.CodeInvalidParameter, "TimeType err")
	}
	flag := service.Device().GetDistributeTableFlagByProductKey(ctx, productKey)
	allTables, err = service.Device().GetAllDistributeTablesByWhere(ctx, &mymodel.SearchDistributeTablesObj{
		Id:            int64(req.DistributeTableId),
		TimeStart:     startTime.Unix(),
		TimeEnd:       endTime.Unix(),
		Flag:          flag,
		DeviceGroupId: deviceInfo0.GroupId,
	})
	if allTables == nil || len(allTables) == 0 {
		return
	}
	db := g.DB(deviceInfo0.DatabaseServer)
	models1 := make([]*gdb.Model, 0) //查总数
	models2 := make([]*gdb.Model, 0) //查列表
	for _, oneTable := range allTables {
		m1 := db.Model(oneTable.Table).Ctx(ctx)
		m2 := db.Model(oneTable.Table).Ctx(ctx)
		// 设备编号查询条件
		m1 = m1.WhereIn(deviceNoTableField, req.DeviceNos)
		m2 = m2.WhereIn(deviceNoTableField, req.DeviceNos)
		//时间条件
		m1 = m1.WhereBetween("sampletime", startTime.UnixMicro(), endTime.UnixMicro())
		m2 = m2.WhereBetween("sampletime", startTime.UnixMicro(), endTime.UnixMicro())
		m2 = m2.Order(order)
		models1 = append(models1, m1)
		models2 = append(models2, m2)
	}

	res.Total, err = db.UnionAll(
		models1...,
	).Count()
	if err != nil {
		return res, err
	}
	if res.Total == 0 {
		return res, nil
	}
	if len(allTables) > 1 && res.Total > maxTotalSearch {
		return res, errors.New("请减少时间范围或者指定查询表去查询,>" + gconv.String(maxTotalSearch))
	}

	switch productKey {
	case consts.ProductKeyM1:
		data := make([]*mymodel.IotDataM1, 0)
		err = db.UnionAll(
			models2...,
		).Offset(req.Offset()).Limit(req.Limit()).Scan(&data)
		if err != nil {
			return
		}
		res.List = arrayUtil.ConvertToAnySlice(data)
	case consts.ProductKeyM2:
		data := make([]*mymodel.IotDataM2, 0)
		err = db.UnionAll(
			models2...,
		).Offset(req.Offset()).Limit(req.Limit()).Scan(&data)
		if err != nil {
			return
		}
		res.List = arrayUtil.ConvertToAnySlice(data)
	case consts.ProductKeyVI1:
		data := make([]*mymodel.IotDataVI1, 0)
		err = db.UnionAll(
			models2...,
		).Offset(req.Offset()).Limit(req.Limit()).Scan(&data)
		if err != nil {
			return
		}
		res.List = arrayUtil.ConvertToAnySlice(data)
	case consts.ProductKeyVI2:
		data := make([]*mymodel.IotDataVI2, 0)
		err = db.UnionAll(
			models2...,
		).Offset(req.Offset()).Limit(req.Limit()).Scan(&data)
		if err != nil {
			return
		}
		res.List = arrayUtil.ConvertToAnySlice(data)
	case consts.ProductKeyM1V1:
		data := make([]*mymodel.IotDataM1VI1, 0)
		err = db.UnionAll(
			models2...,
		).Offset(req.Offset()).Limit(req.Limit()).Scan(&data)
		if err != nil {
			return
		}
		res.List = arrayUtil.ConvertToAnySlice(data)
	case consts.ProductKeyM1V2:
		data := make([]*mymodel.IotDataM1VI2, 0)
		err = db.UnionAll(
			models2...,
		).Offset(req.Offset()).Limit(req.Limit()).Scan(&data)
		if err != nil {
			return
		}
		res.List = arrayUtil.ConvertToAnySlice(data)
	case consts.ProductKeyM2V1:
		data := make([]*mymodel.IotDataM2VI1, 0)
		err = db.UnionAll(
			models2...,
		).Offset(req.Offset()).Limit(req.Limit()).Scan(&data)
		if err != nil {
			return
		}
		res.List = arrayUtil.ConvertToAnySlice(data)
	case consts.ProductKeyM2V2:
		data := make([]*mymodel.IotDataM2VI2, 0)
		err = db.UnionAll(
			models2...,
		).Offset(req.Offset()).Limit(req.Limit()).Scan(&data)
		if err != nil {
			return
		}
		res.List = arrayUtil.ConvertToAnySlice(data)
	default:
		return nil, errors.New("invalid product key")
	}
	return res, nil
}

func (s *sIotdata) SummaryPageLst(ctx context.Context, req *v1.IotDataSummaryPageLstReq) (res *v1.IotDataSummaryPageLstRes, err error) {
	res = &v1.IotDataSummaryPageLstRes{
		PageLstRes: &api.PageLstRes[any]{
			List:  make([]any, 0),
			Total: 0,
		},
	}
	var (
		//order              = "sampletime desc"
		allTables          []*mymodel.DistributeTableResponse
		deviceNoTableField = dao.IotAeParamDemo.Columns().DeviceNo
		startTime          time.Time
		endTime            time.Time
	)
	//if req.OrderBy != "" {
	//	order = req.OrderBy
	//}

	// 参数校验增强
	if len(req.DeviceNos) == 0 {
		return nil, gerror.NewCode(gcode.CodeInvalidParameter, "deviceNo required")
	}
	// 维度校验
	if req.Dimension != 2 && req.Dimension != 3 {
		return nil, gerror.NewCode(gcode.CodeInvalidParameter, "dimension must be 2 or 3")
	}
	// X轴和Y轴参数校验
	if req.XAxisParam == "" || req.YAxisParam == "" {
		return nil, gerror.NewCode(gcode.CodeInvalidParameter, "XAxisParam and YAxisParam are required")
	}
	// 3D模式下Z轴参数校验
	if req.Dimension == 3 && req.ZAxisParam == "" {
		return nil, gerror.NewCode(gcode.CodeInvalidParameter, "ZAxisParam is required for 3D dimension")
	}

	// 将驼峰命名转换为蛇形命名
	convertToSnakeCase := func(s string) string {
		var result []rune
		for i, r := range s {
			if i > 0 && unicode.IsUpper(r) {
				result = append(result, '_')
			}
			result = append(result, unicode.ToLower(r))
		}
		return string(result)
	}

	xAxisField := convertToSnakeCase(req.XAxisParam)
	yAxisField := convertToSnakeCase(req.YAxisParam)
	var zAxisField string
	if req.Dimension == 3 {
		zAxisField = convertToSnakeCase(req.ZAxisParam)
	}

	deviceInfo0, err := service.Device().GetByDeviceNo(ctx, req.DeviceNos[0])
	if err != nil || deviceInfo0 == nil || deviceInfo0.Id == 0 {
		return nil, gerror.NewCode(gcode.CodeInvalidParameter, "deviceNo not exist")
	}

	isSuperDept, currentUser, _ := service.User().IsCurrentUserSuperDept(ctx)
	if !isSuperDept && currentUser.DeptId != int(deviceInfo0.DeptId) {
		return nil, gerror.NewCode(gcode.CodeSecurityReason, "permission deny")
	}

	productInfo, err := service.Product().GetByProductId(ctx, deviceInfo0.ProductId)
	if err != nil || productInfo == nil || productInfo.Id == 0 {
		return nil, gerror.NewCode(gcode.CodeInvalidParameter, "productInfo not exist")
	}
	productKey := productInfo.ProductKey

	// 时间条件处理优化
	switch req.TimeType {
	case 0:
		if len(req.TimeRange) != 2 {
			return nil, gerror.NewCode(gcode.CodeInvalidParameter, "时间范围参数格式错误")
		}
		// 新增时间转换逻辑
		startTime, err = time.Parse(time.RFC3339, req.TimeRange[0])
		if err != nil {
			return nil, gerror.WrapCode(gcode.CodeInvalidParameter, err, "开始时间格式错误")
		}
		endTime, err = time.Parse(time.RFC3339, req.TimeRange[1])
		if err != nil {
			return nil, gerror.WrapCode(gcode.CodeInvalidParameter, err, "结束时间格式错误")
		}
	case 1:
		startTime = gtime.Now().Add(-time.Duration(req.Recent) * time.Second).Time
		endTime = gtime.Now().Time
	default:
		return nil, gerror.NewCode(gcode.CodeInvalidParameter, "TimeType err")
	}

	flag := service.Device().GetDistributeTableFlagByProductKey(ctx, productKey)
	allTables, err = service.Device().GetAllDistributeTablesByWhere(ctx, &mymodel.SearchDistributeTablesObj{
		Id:            int64(req.DistributeTableId),
		TimeStart:     startTime.Unix(),
		TimeEnd:       endTime.Unix(),
		Flag:          flag,
		DeviceGroupId: deviceInfo0.GroupId,
	})
	if allTables == nil || len(allTables) == 0 {
		return
	}

	db := g.DB(deviceInfo0.DatabaseServer)
	models1 := make([]*gdb.Model, 0) //查总数
	models2 := make([]*gdb.Model, 0) //查列表

	// 构建查询字段列表
	fields := []string{"device_no", xAxisField, yAxisField}
	if req.Dimension == 3 {
		fields = append(fields, zAxisField)
	}

	for _, oneTable := range allTables {
		m1 := db.Model(oneTable.Table).Ctx(ctx)
		m2 := db.Model(oneTable.Table).Ctx(ctx)

		// 选择需要的字段
		m1 = m1.Fields(fields)
		m2 = m2.Fields(fields)

		// 设备编号查询条件
		m1 = m1.WhereIn(deviceNoTableField, req.DeviceNos)
		m2 = m2.WhereIn(deviceNoTableField, req.DeviceNos)

		// 时间条件
		m1 = m1.WhereBetween("sampletime", startTime.UnixMicro(), endTime.UnixMicro())
		m2 = m2.WhereBetween("sampletime", startTime.UnixMicro(), endTime.UnixMicro())

		// 自定义过滤条件
		if req.XFilterMode == "custom" {
			m1 = m1.WhereBetween(xAxisField, req.XMin, req.XMax)
			m2 = m2.WhereBetween(xAxisField, req.XMin, req.XMax)
		}
		if req.YFilterMode == "custom" {
			m1 = m1.WhereBetween(yAxisField, req.YMin, req.YMax)
			m2 = m2.WhereBetween(yAxisField, req.YMin, req.YMax)
		}
		if req.Dimension == 3 && req.ZFilterMode == "custom" {
			m1 = m1.WhereBetween(zAxisField, req.ZMin, req.ZMax)
			m2 = m2.WhereBetween(zAxisField, req.ZMin, req.ZMax)
		}

		//m2 = m2.Order(order)
		models1 = append(models1, m1)
		models2 = append(models2, m2)
	}

	// 查询总数
	res.Total, err = db.UnionAll(
		models1...,
	).Count()
	if err != nil {
		return res, err
	}
	if res.Total == 0 {
		return res, nil
	}
	if len(allTables) > 1 && res.Total > maxTotalSearch {
		return res, errors.New("请减少时间范围或者指定查询表去查询,>" + gconv.String(maxTotalSearch))
	}

	// 查询数据
	var resultData gdb.Result
	resultData, err = db.UnionAll(
		models2...,
	).Offset(req.Offset()).Limit(req.Limit()).All()
	if err != nil {
		return res, err
	}
	// 转换为要求的数组格式
	for _, item := range resultData {
		var row []interface{}
		row = append(row, item["device_no"])
		// 添加X轴值
		row = append(row, item[xAxisField])
		// 添加Y轴值
		row = append(row, item[yAxisField])
		// 如果是3D模式，添加Z轴值
		if req.Dimension == 3 {
			row = append(row, item[zAxisField])
		}
		res.List = append(res.List, row)
	}

	return res, nil
}

// GetOne 获取单条物联网数据
func (s *sIotdata) GetOne(ctx context.Context, req *v1.IotDataGetOneReq) (res *v1.IotDataGetOneRes, err error) {
	res = new(v1.IotDataGetOneRes)
	sec := req.Sampletime / 1000000
	deviceNoTableField := "device_no"
	deviceInfo0, err := service.Device().GetByDeviceNo(ctx, req.DeviceNo)
	if err != nil || deviceInfo0 == nil || deviceInfo0.Id == 0 {
		return nil, gerror.NewCode(gcode.CodeInvalidParameter, "deviceNo not exist")
	}
	isSuperDept, currentUser, _ := service.User().IsCurrentUserSuperDept(ctx)
	if !isSuperDept && currentUser.DeptId != int(deviceInfo0.DeptId) {
		return nil, gerror.NewCode(gcode.CodeSecurityReason, "permission deny")
	}
	productInfo, err := service.Product().GetByProductId(ctx, deviceInfo0.ProductId)
	if err != nil || productInfo == nil || productInfo.Id == 0 {
		return nil, gerror.NewCode(gcode.CodeInvalidParameter, "productInfo not exist")
	}
	productKey := productInfo.ProductKey
	flag := service.Device().GetDistributeTableFlagByProductKey(ctx, productKey)
	allTables, err := service.Device().GetAllDistributeTablesByWhere(ctx, &mymodel.SearchDistributeTablesObj{
		TimeStart:     sec,
		TimeEnd:       sec,
		Flag:          flag,
		DeviceGroupId: deviceInfo0.GroupId,
	})
	if allTables == nil || len(allTables) == 0 {
		return
	}
	db := g.DB(deviceInfo0.DatabaseServer)
	models := make([]*gdb.Model, 0)
	for _, oneTable := range allTables {
		m := db.Model(oneTable.Table).Ctx(ctx)
		// 设备编号查询条件
		m = m.Where(deviceNoTableField, req.DeviceNo)
		m = m.Where("sampletime", req.Sampletime)
		models = append(models, m)
	}
	switch productKey {
	case consts.ProductKeyM1:
		var data *mymodel.IotDataM1
		err = db.UnionAll(
			models...,
		).Scan(&data)
		if err != nil {
			return
		}
		res.DeviceNo = data.DeviceNo
		res.Sampletime = int64(data.Sampletime)
		res.Data = data
	case consts.ProductKeyM2:
		var data *mymodel.IotDataM2
		err = db.UnionAll(
			models...,
		).Scan(&data)
		if err != nil {
			return
		}
		res.DeviceNo = data.DeviceNo
		res.Sampletime = int64(data.Sampletime)
		res.Data = data
	case consts.ProductKeyVI1:
		var data *mymodel.IotDataVI1
		err = db.UnionAll(
			models...,
		).Scan(&data)
		if err != nil {
			return
		}
		res.DeviceNo = data.DeviceNo
		res.Sampletime = int64(data.Sampletime)
		res.Data = data
	case consts.ProductKeyVI2:
		var data *mymodel.IotDataVI2
		err = db.UnionAll(
			models...,
		).Scan(&data)
		if err != nil {
			return
		}
		res.DeviceNo = data.DeviceNo
		res.Sampletime = int64(data.Sampletime)
		res.Data = data
	case consts.ProductKeyM1V1:
		var data *mymodel.IotDataM1VI1
		err = db.UnionAll(
			models...,
		).Scan(&data)
		if err != nil {
			return
		}
		res.DeviceNo = data.DeviceNo
		res.Sampletime = int64(data.Sampletime)
		res.Data = data
	case consts.ProductKeyM1V2:
		var data *mymodel.IotDataM1VI2
		err = db.UnionAll(
			models...,
		).Scan(&data)
		if err != nil {
			return
		}
		res.DeviceNo = data.DeviceNo
		res.Sampletime = int64(data.Sampletime)
		res.Data = data
	case consts.ProductKeyM2V1:
		var data *mymodel.IotDataM2VI1
		err = db.UnionAll(
			models...,
		).Scan(&data)
		if err != nil {
			return
		}
		res.DeviceNo = data.DeviceNo
		res.Sampletime = int64(data.Sampletime)
		res.Data = data
	case consts.ProductKeyM2V2:
		var data *mymodel.IotDataM2VI2
		err = db.UnionAll(
			models...,
		).Scan(&data)
		if err != nil {
			return
		}
		res.DeviceNo = data.DeviceNo
		res.Sampletime = int64(data.Sampletime)
		res.Data = data
	default:
		return nil, errors.New("invalid product key")
	}
	return
}

func (s *sIotdata) GetOneWave(ctx context.Context, req *v1.IotDataGetOneWaveReq) (res *v1.IotDataGetOneWaveRes, err error) {
	typ := req.Type
	res = new(v1.IotDataGetOneWaveRes)
	sec := req.Sampletime / 1000000
	deviceNoTableField := "device_no"
	deviceInfo0, err := service.Device().GetByDeviceNo(ctx, req.DeviceNo)
	if err != nil || deviceInfo0 == nil || deviceInfo0.Id == 0 {
		return nil, gerror.NewCode(gcode.CodeInvalidParameter, "deviceNo not exist")
	}
	isSuperDept, currentUser, _ := service.User().IsCurrentUserSuperDept(ctx)
	if !isSuperDept && currentUser.DeptId != int(deviceInfo0.DeptId) {
		return nil, gerror.NewCode(gcode.CodeSecurityReason, "permission deny")
	}
	flag := consts.AeDistributeTableFlagNone
	if typ == consts.WaveTypeAe {
		flag = consts.AeDistributeTableFlagAeWave
	} else if typ == consts.WaveTypeVi {
		flag = consts.AeDistributeTableFlagViwave
	} else {
		return nil, gerror.NewCode(gcode.CodeInvalidParameter, "wave type not exist")
	}
	allTables, err := service.Device().GetAllDistributeTablesByWhere(ctx, &mymodel.SearchDistributeTablesObj{
		TimeStart:     sec,
		TimeEnd:       sec,
		Flag:          flag,
		DeviceGroupId: deviceInfo0.GroupId,
	})
	if allTables == nil || len(allTables) != 1 {
		err = gerror.NewCode(gcode.CodeInvalidParameter, "get all tables failed")
		return
	}
	oneTable := allTables[0]
	m := g.DB(oneTable.ServerKey).Model(oneTable.Table).Ctx(ctx)
	// 设备编号查询条件
	m = m.Where(deviceNoTableField, req.DeviceNo)

	if typ == consts.WaveTypeAe {
		var data *mymodel.IotWaveAe
		if req.PageFlag == -1 {
			m = m.Where("sampletime <", req.Sampletime).OrderDesc("sampletime")
			err = m.Scan(&data)
			if err != nil {
				return nil, err
			}
			if data == nil || data.DeviceNo != req.DeviceNo {
				// 不存在则去前一个表去找
				var distributeTableRow *entity.AppDistributeTables
				err = dao.AppDistributeTables.Ctx(ctx).Where("id <", oneTable.Id).OrderDesc("id").Scan(&distributeTableRow)
				if err != nil || distributeTableRow == nil || distributeTableRow.Id == 0 {
					// 说明真找不到了
					return
				}
				err = g.DB(distributeTableRow.ServerKey).Model(distributeTableRow.Table).Ctx(ctx).
					Where(deviceNoTableField, req.DeviceNo).
					Where("sampletime <", req.Sampletime).
					OrderDesc("sampletime").Scan(&data)
				if err != nil {
					return nil, err
				}
			}
		} else if req.PageFlag == 1 {
			m = m.Where("sampletime >", req.Sampletime).OrderAsc("sampletime")
			err = m.Scan(&data)
			if err != nil {
				return
			}
			if data == nil || data.DeviceNo != req.DeviceNo {
				// 不存在则去后一个表去找
				var distributeTableRow *entity.AppDistributeTables
				err = dao.AppDistributeTables.Ctx(ctx).Where("id >", oneTable.Id).OrderAsc("id").Scan(&distributeTableRow)
				if err != nil || distributeTableRow == nil || distributeTableRow.Id == 0 {
					// 说明真找不到了
					return
				}
				err = g.DB(distributeTableRow.ServerKey).Model(distributeTableRow.Table).Ctx(ctx).
					Where(deviceNoTableField, req.DeviceNo).
					Where("sampletime >", req.Sampletime).
					OrderAsc("sampletime").Scan(&data)
				if err != nil {
					return nil, err
				}
			}
		} else {
			m = m.Where("sampletime", req.Sampletime)
			err = m.Scan(&data)
			if err != nil {
				return
			}
		}

		if data == nil || data.DeviceNo != req.DeviceNo {
			return
		}
		paramRes, err := s.GetOne(ctx, &v1.IotDataGetOneReq{
			Sampletime: int64(data.Sampletime),
			DeviceNo:   data.DeviceNo,
		})
		if err != nil {
			return nil, err
		}
		var bytes []byte
		if data.Flag == consts.StoreFlagOss {
			minioconfkey := minioUtil.GetMinioConfkeyByServername(ctx, data.StoreServer)
			bytes, _ = minioUtil.GetBytes(ctx, minioconfkey, data.Bucket, data.Path)
		} else {
			filePath, _ := waveFileSave.GetWaveSavePath(data.Bucket, data.Path)
			bytes, _ = waveFileSave.ReadFileData(filePath)
		}
		res.Data = paramRes
		res.Wave = hex.EncodeToString(bytes)
	} else if typ == consts.WaveTypeVi {
		var data *mymodel.IotWaveVi
		if req.PageFlag == -1 {
			m = m.Where("sampletime <", req.Sampletime).OrderDesc("sampletime")
			err = m.Scan(&data)
			if err != nil {
				return nil, err
			}
			if data == nil || data.DeviceNo != req.DeviceNo {
				// 不存在则去前一个表去找
				var distributeTableRow *entity.AppDistributeTables
				err = dao.AppDistributeTables.Ctx(ctx).Where("id <", oneTable.Id).OrderDesc("id").Scan(&distributeTableRow)
				if err != nil || distributeTableRow == nil || distributeTableRow.Id == 0 {
					// 说明真找不到了
					return
				}
				err = g.DB(distributeTableRow.ServerKey).Model(distributeTableRow.Table).Ctx(ctx).
					Where(deviceNoTableField, req.DeviceNo).
					Where("sampletime <", req.Sampletime).
					OrderDesc("sampletime").Scan(&data)
				if err != nil {
					return nil, err
				}
			}
		} else if req.PageFlag == 1 {
			m = m.Where("sampletime >", req.Sampletime).OrderAsc("sampletime")
			err = m.Scan(&data)
			if err != nil {
				return
			}
			if data == nil || data.DeviceNo != req.DeviceNo {
				// 不存在则去后一个表去找
				var distributeTableRow *entity.AppDistributeTables
				err = dao.AppDistributeTables.Ctx(ctx).Where("id >", oneTable.Id).OrderAsc("id").Scan(&distributeTableRow)
				if err != nil || distributeTableRow == nil || distributeTableRow.Id == 0 {
					// 说明真找不到了
					return
				}
				err = g.DB(distributeTableRow.ServerKey).Model(distributeTableRow.Table).Ctx(ctx).
					Where(deviceNoTableField, req.DeviceNo).
					Where("sampletime >", req.Sampletime).
					OrderAsc("sampletime").Scan(&data)
				if err != nil {
					return nil, err
				}
			}
		} else {
			m = m.Where("sampletime", req.Sampletime)
			err = m.Scan(&data)
			if err != nil {
				return
			}
		}

		if data == nil || data.DeviceNo != req.DeviceNo {
			return
		}
		paramRes, err := s.GetOne(ctx, &v1.IotDataGetOneReq{
			Sampletime: int64(data.Sampletime),
			DeviceNo:   data.DeviceNo,
		})
		if err != nil {
			return nil, err
		}
		var bytes []byte
		if data.Flag == consts.StoreFlagOss {
			minioconfkey := minioUtil.GetMinioConfkeyByServername(ctx, data.StoreServer)
			bytes, _ = minioUtil.GetBytes(ctx, minioconfkey, data.Bucket, data.Path)
		} else {
			filePath, _ := waveFileSave.GetWaveSavePath(data.Bucket, data.Path)
			bytes, _ = waveFileSave.ReadFileData(filePath)
		}
		res.Data = paramRes.Data
		res.Wave = hex.EncodeToString(bytes)
	}
	return
}

// GetRealtimeWave 从缓存获取
func (s *sIotdata) GetRealtimeWave(ctx context.Context, req *v1.IotDataGetRealtimeWaveReq) (res *v1.IotDataGetRealtimeWaveRes, err error) {
	typ := req.Type
	res = new(v1.IotDataGetRealtimeWaveRes)
	deviceInfo0, err := service.Device().GetByDeviceNo(ctx, req.DeviceNo)
	if err != nil || deviceInfo0 == nil || deviceInfo0.Id == 0 {
		return nil, gerror.NewCode(gcode.CodeInvalidParameter, "deviceNo not exist")
	}
	isSuperDept, currentUser, _ := service.User().IsCurrentUserSuperDept(ctx)
	if !isSuperDept && currentUser.DeptId != int(deviceInfo0.DeptId) {
		return nil, gerror.NewCode(gcode.CodeSecurityReason, "permission deny")
	}
	productInfo, err := service.Product().GetByProductId(ctx, deviceInfo0.ProductId)
	if err != nil || productInfo == nil || productInfo.Id == 0 {
		return nil, gerror.NewCode(gcode.CodeInvalidParameter, "productInfo not exist")
	}
	productKey := productInfo.ProductKey

	flag := consts.AeDistributeTableFlagNone //波形的
	if typ == consts.WaveTypeAe {
		flag = consts.AeDistributeTableFlagAeWave
	} else if typ == consts.WaveTypeVi {
		flag = consts.AeDistributeTableFlagViwave
	} else {
		return nil, gerror.NewCode(gcode.CodeInvalidParameter, "wave type not exist")
	}

	flag1 := consts.AeDistributeTableFlagNone //参数的
	if productKey == consts.ProductKeyM1 {
		flag1 = consts.AeDistributeTableFlagAeParam
	} else if productKey == consts.ProductKeyM2 {
		flag1 = consts.AeDistributeTableFlagAeParam
	} else if productKey == consts.ProductKeyVI1 {
		flag1 = consts.AeDistributeTableFlagViParam
	} else if productKey == consts.ProductKeyVI2 {
		flag1 = consts.AeDistributeTableFlagViParam
	} else if productKey == consts.ProductKeyM1V1 {
		flag1 = consts.AeDistributeTableFlagAeViParam
	} else if productKey == consts.ProductKeyM1V2 {
		flag1 = consts.AeDistributeTableFlagAeViParam
	} else if productKey == consts.ProductKeyM2V1 {
		flag1 = consts.AeDistributeTableFlagAeViParam
	} else if productKey == consts.ProductKeyM2V2 {
		flag1 = consts.AeDistributeTableFlagAeViParam
	} else {
		return nil, gerror.NewCode(gcode.CodeInvalidParameter, "NOT MATCH PRODUCT PARAM DATA")
	}

	formatType := service.Device().GetAeDataCacheFormatType(ctx, int(req.Domain), int(req.Direction))
	cacheKey := cacheConsts.GetDeviceDataKey(req.DeviceNo, flag, formatType)
	wavaCacheD := &mymodel.WaveCacheData{}
	// 从缓存获取
	if v, _ := service.GetCacheInstance().Get(ctx, cacheKey); !v.IsEmpty() {
		err = v.Struct(wavaCacheD)
		if err != nil {
			return
		}
	}
	if productKey == consts.ProductKeyM1 {
		dataParam := &mymodel.IotDataM1{}
		formatType1 := service.Device().GetAeDataCacheFormatType(ctx, 0, 0)
		cacheKey1 := cacheConsts.GetDeviceDataKey(req.DeviceNo, flag1, formatType1)
		// 从缓存获取
		if v, _ := service.GetCacheInstance().Get(ctx, cacheKey1); !v.IsEmpty() {
			err = v.Struct(dataParam)
			if err != nil {
				return
			}
		}
		res.Data = dataParam
		if dataParam.Sampletime == wavaCacheD.Sampletime {
			res.Wave = wavaCacheD.Wave
		}
	} else if productKey == consts.ProductKeyM2 {
		dataParam := &mymodel.IotDataM2{}
		formatType1 := service.Device().GetAeDataCacheFormatType(ctx, 0, 0)
		cacheKey1 := cacheConsts.GetDeviceDataKey(req.DeviceNo, flag1, formatType1)
		// 从缓存获取
		if v, _ := service.GetCacheInstance().Get(ctx, cacheKey1); !v.IsEmpty() {
			err = v.Struct(dataParam)
			if err != nil {
				return
			}
		}
		res.Data = dataParam
		if dataParam.Sampletime == wavaCacheD.Sampletime {
			res.Wave = wavaCacheD.Wave
		}
	} else if productKey == consts.ProductKeyVI1 {
		dataParam := &mymodel.IotDataVI1{}
		formatType1 := service.Device().GetAeDataCacheFormatType(ctx, 0, 0)
		cacheKey1 := cacheConsts.GetDeviceDataKey(req.DeviceNo, flag1, formatType1)
		// 从缓存获取
		if v, _ := service.GetCacheInstance().Get(ctx, cacheKey1); !v.IsEmpty() {
			err = v.Struct(dataParam)
			if err != nil {
				return
			}
		}
		res.Data = dataParam
		if dataParam.Sampletime == wavaCacheD.Sampletime {
			res.Wave = wavaCacheD.Wave
		}
	} else if productKey == consts.ProductKeyVI2 {
		dataParam := &mymodel.IotDataVI2{}
		formatType1 := service.Device().GetAeDataCacheFormatType(ctx, 0, 0)
		cacheKey1 := cacheConsts.GetDeviceDataKey(req.DeviceNo, flag1, formatType1)
		// 从缓存获取
		if v, _ := service.GetCacheInstance().Get(ctx, cacheKey1); !v.IsEmpty() {
			err = v.Struct(dataParam)
			if err != nil {
				return
			}
		}
		res.Data = dataParam
		if dataParam.Sampletime == wavaCacheD.Sampletime {
			res.Wave = wavaCacheD.Wave
		}
	} else if productKey == consts.ProductKeyM1V1 {
		dataParam := &mymodel.IotDataM1VI1{}
		formatType1 := service.Device().GetAeDataCacheFormatType(ctx, 0, 0)
		cacheKey1 := cacheConsts.GetDeviceDataKey(req.DeviceNo, flag1, formatType1)
		// 从缓存获取
		if v, _ := service.GetCacheInstance().Get(ctx, cacheKey1); !v.IsEmpty() {
			err = v.Struct(dataParam)
			if err != nil {
				return
			}
		}
		res.Data = dataParam
		if dataParam.Sampletime == wavaCacheD.Sampletime {
			res.Wave = wavaCacheD.Wave
		}
	} else if productKey == consts.ProductKeyM1V2 {
		dataParam := &mymodel.IotDataM1VI2{}
		formatType1 := service.Device().GetAeDataCacheFormatType(ctx, 0, 0)
		cacheKey1 := cacheConsts.GetDeviceDataKey(req.DeviceNo, flag1, formatType1)
		// 从缓存获取
		if v, _ := service.GetCacheInstance().Get(ctx, cacheKey1); !v.IsEmpty() {
			err = v.Struct(dataParam)
			if err != nil {
				return
			}
		}
		res.Data = dataParam
		if dataParam.Sampletime == wavaCacheD.Sampletime {
			res.Wave = wavaCacheD.Wave
		}
	} else if productKey == consts.ProductKeyM2V1 {
		dataParam := &mymodel.IotDataM2VI1{}
		formatType1 := service.Device().GetAeDataCacheFormatType(ctx, 0, 0)
		cacheKey1 := cacheConsts.GetDeviceDataKey(req.DeviceNo, flag1, formatType1)
		// 从缓存获取
		if v, _ := service.GetCacheInstance().Get(ctx, cacheKey1); !v.IsEmpty() {
			err = v.Struct(dataParam)
			if err != nil {
				return
			}
		}
		res.Data = dataParam
		if dataParam.Sampletime == wavaCacheD.Sampletime {
			res.Wave = wavaCacheD.Wave
		}
	} else {
		dataParam := &mymodel.IotDataM2VI2{}
		formatType1 := service.Device().GetAeDataCacheFormatType(ctx, 0, 0)
		cacheKey1 := cacheConsts.GetDeviceDataKey(req.DeviceNo, flag1, formatType1)
		// 从缓存获取
		if v, _ := service.GetCacheInstance().Get(ctx, cacheKey1); !v.IsEmpty() {
			err = v.Struct(dataParam)
			if err != nil {
				return
			}
		}
		res.Data = dataParam
		if dataParam.Sampletime == wavaCacheD.Sampletime {
			res.Wave = wavaCacheD.Wave
		}
	}
	return
}

// Del 批量删除物联网数据
func (s *sIotdata) Del(ctx context.Context, req *v1.IotDataDelReq) error {
	// 不支持这种删法 太消耗性能
	return nil
}

// DelByDeviceNo 删除该设备所有物联网数据
func (s *sIotdata) DelByDeviceNo(ctx context.Context, deviceNo string) (err error) {
	deviceInfo, _ := service.Device().GetByDeviceNo(ctx, deviceNo)
	if deviceInfo == nil || deviceInfo.Id == 0 {
		return nil
	}
	var distributeTableRows []*entity.AppDistributeTables
	err = dao.AppDistributeTables.Ctx(ctx).Where(dao.AppDistributeTables.Columns().DeviceGroupId, deviceInfo.GroupId).Scan(&distributeTableRows)
	if err != nil || len(distributeTableRows) == 0 {
		return nil
	}
	for _, distributeTableRow := range distributeTableRows {
		offset := 0
		limit := 20000
		if distributeTableRow.Flag == consts.AeDistributeTableFlagAeWave {
			for {
				var datas []*entity.IotAeWaveDemo
				err = g.DB(distributeTableRow.ServerKey).Model(distributeTableRow.Table).
					Where(dao.IotAeWaveDemo.Columns().DeviceNo, deviceNo).
					Offset(offset).Limit(limit).Scan(&datas)
				for _, item := range datas {
					flag := waveFileSave.GetSaveFlag(ctx)
					if flag == consts.StoreFlagOss {
						minioconf := minioUtil.GetMinioConfigByServername(ctx, item.StoreServer)
						if minioconf == nil || minioconf.Bucket == "" {
							return errors.New("minio minio config err")
						}
						err = service.Aeemqx().DeleteWaveFromOss(ctx, item.StoreServer, item.Bucket, item.Path)
						if err != nil {
							return err
						}
					} else {
						fullpath, _ := waveFileSave.GetWaveSavePath(item.Bucket, item.Path)
						_ = waveFileSave.DeleteFile(fullpath)
					}
				}
				offset = offset + limit
				if datas == nil || len(datas) < limit {
					break
				}
			}
		} else if distributeTableRow.Flag == consts.AeDistributeTableFlagViwave {
			for {
				var datas []*entity.IotViWaveDemo
				err = g.DB(distributeTableRow.ServerKey).Model(distributeTableRow.Table).
					Where(dao.IotViWaveDemo.Columns().DeviceNo, deviceNo).
					Offset(offset).Limit(limit).Scan(&datas)
				for _, item := range datas {
					flag := waveFileSave.GetSaveFlag(ctx)
					if flag == consts.StoreFlagOss {
						minioconf := minioUtil.GetMinioConfigByServername(ctx, item.StoreServer)
						if minioconf == nil || minioconf.Bucket == "" {
							return errors.New("minio minio config err")
						}
						err = service.Aeemqx().DeleteWaveFromOss(ctx, item.StoreServer, item.Bucket, item.Path)
						if err != nil {
							return err
						}
					} else {
						fullpath, _ := waveFileSave.GetWaveSavePath(item.Bucket, item.Path)
						_ = waveFileSave.DeleteFile(fullpath)
					}
				}
				offset = offset + limit
				if datas == nil || len(datas) < limit {
					break
				}
			}
		}
		_, err = g.DB(distributeTableRow.ServerKey).Model(distributeTableRow.Table).
			Where("device_no", deviceNo).Delete()
	}
	return
}

func (s *sIotdata) DelByDistributeTableId(ctx context.Context, distributeId int64) (err error) {
	var distributeTableRow *entity.AppDistributeTables
	err = dao.AppDistributeTables.Ctx(ctx).Where("id", distributeId).Scan(&distributeTableRow)
	if err != nil {
		return
	}
	if distributeTableRow == nil || distributeTableRow.Id == 0 {
		return errors.New("distribute table id is 0")
	}
	offset := 0
	limit := 20000
	// 遍历循环删除  波形除了表还有波形本身文件
	if distributeTableRow.Flag == consts.AeDistributeTableFlagAeWave {
		for {
			var datas []*entity.IotAeWaveDemo
			err = g.DB(distributeTableRow.ServerKey).Model(distributeTableRow.Table).Offset(offset).Limit(limit).Scan(&datas)
			for _, item := range datas {
				flag := waveFileSave.GetSaveFlag(ctx)
				if flag == consts.StoreFlagOss {
					minioconf := minioUtil.GetMinioConfigByServername(ctx, item.StoreServer)
					if minioconf == nil || minioconf.Bucket == "" {
						return errors.New("minio minio config err")
					}
					err = service.Aeemqx().DeleteWaveFromOss(ctx, item.StoreServer, item.Bucket, item.Path)
					if err != nil {
						return err
					}
				} else {
					fullpath, _ := waveFileSave.GetWaveSavePath(item.Bucket, item.Path)
					_ = waveFileSave.DeleteFile(fullpath)
				}
			}
			offset = offset + limit
			if datas == nil || len(datas) < limit {
				break
			}
		}
	} else if distributeTableRow.Flag == consts.AeDistributeTableFlagViwave {
		for {
			var datas []*entity.IotViWaveDemo
			err = g.DB(distributeTableRow.ServerKey).Model(distributeTableRow.Table).Offset(offset).Limit(limit).Scan(&datas)
			for _, item := range datas {
				flag := waveFileSave.GetSaveFlag(ctx)
				if flag == consts.StoreFlagOss {
					minioconf := minioUtil.GetMinioConfigByServername(ctx, item.StoreServer)
					if minioconf == nil || minioconf.Bucket == "" {
						return errors.New("minio minio config err")
					}
					err = service.Aeemqx().DeleteWaveFromOss(ctx, item.StoreServer, item.Bucket, item.Path)
					if err != nil {
						return err
					}
				} else {
					fullpath, _ := waveFileSave.GetWaveSavePath(item.Bucket, item.Path)
					_ = waveFileSave.DeleteFile(fullpath)
				}
			}
			offset = offset + limit
			if datas == nil || len(datas) < limit {
				break
			}
		}
	}
	err = dbUtil.TruncateTable(ctx, distributeTableRow.ServerKey, distributeTableRow.Table)
	if err != nil {
		return err
	}
	return
}
